package cn.learn.server;

import java.io.IOException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class SyncRequestProcessor extends Thread implements RequestProcessor{
    private final ZooKeeperServer zks;
    private final BlockingQueue<Request> queuedRequests = new LinkedBlockingQueue<Request>();

    private final RequestProcessor nextProcessor;
    public SyncRequestProcessor(ZooKeeperServer zks, RequestProcessor nextProcessor) {
        this.zks = zks;
        this.nextProcessor = nextProcessor;
    }

    @Override
    public void run() {
        try {
            while (true) {
                System.out.println("Sync处理器的阻塞队列。。。");
                // 这里原本是轮询，查每个请求是否1/2+1个节点同步数据，现在改成直接通过
                Request request = queuedRequests.take();

                if (zks.getZKDatabase().append(request)){
                    new Thread(new Runnable() {
                        @Override
                        public void run() {
                            try {
                                zks.takeSnapshot();
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                    }).start();
                }

                nextProcessor.processRequest(request);
            }
        }catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void processRequest(Request request) throws RequestProcessorException {
        System.out.println("同步处理器开始处理。。。");
        queuedRequests.add(request);
    }

    @Override
    public void shutdown() {

    }
}
